package com.platform.common.redis.hash;

import com.platform.common.redis.RedisService;
import com.platform.common.utils.SerializeUtil;
import com.platform.common.utils.SpringUtil;
import org.apache.commons.lang3.NotImplementedException;
import org.apache.commons.lang3.math.NumberUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import redis.clients.jedis.*;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * 对redis pool进行hash分片
 *
 * @author LiangMingHui
 *
 */
public class HashRedisUtil implements RedisService {

	private static final Logger logger = LoggerFactory.getLogger(HashRedisUtil.class);

	private static final String DEFAULT_REDIS_SEPARATOR = ";";

	private static final String HOST_PORT_SEPARATOR = ":";

	private JedisPool[] jedisPools = new JedisPool[0];

	private static final HashRedisUtil instance = new HashRedisUtil();

	public static HashRedisUtil getInstance() {
		return instance;
	}

	private HashRedisUtil() {
		try {
			// 初始化连接池
			jedisPools = (JedisPool[]) SpringUtil.getBeanByClass(JedisPool[].class);
			logger.info("初始化jedisPools成功");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}


	/**
	 * 实现jedis连接的获取和释放，具体的redis业务逻辑由executor实现
	 *
	 * @param executor
	 *            RedisExecutor接口的实现类
	 * @return
	 */
	public <T> T execute(String key, HashRedisExecutor<T> executor) {
		Jedis jedis = jedisPools[(0x7FFFFFFF & key.hashCode()) % jedisPools.length].getResource();
		T result = null;
		try {
			result = executor.execute(jedis);
		} finally {
			if (jedis != null) {
				jedis.close();
			}
		}
		return result;
	}

	public String set(final String key, final String value) {
		return execute(key, new HashRedisExecutor<String>() {
			@Override
			public String execute(Jedis jedis) {
				return jedis.set(key, value);
			}
		});
	}

	public String set(final String key, final String value, final String nxxx, final String expx, final long time) {
		return execute(key, new HashRedisExecutor<String>() {
			@Override
			public String execute(Jedis jedis) {
				return jedis.set(key, value, nxxx, expx, time);
			}
		});
	}

	public String get(final String key) {
		return execute(key, new HashRedisExecutor<String>() {
			@Override
			public String execute(Jedis jedis) {
				return jedis.get(key);
			}
		});
	}

	public Boolean exists(final String key) {
		return execute(key, new HashRedisExecutor<Boolean>() {
			@Override
			public Boolean execute(Jedis jedis) {
				return jedis.exists(key);
			}
		});
	}

	public Long setnx(final String key, final String value) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				Long setnx = jedis.setnx(key, value);
				return setnx;
			}
		});
	}

	public String setex(final String key, final int seconds, final String value) {
		return execute(key, new HashRedisExecutor<String>() {
			@Override
			public String execute(Jedis jedis) {
				return jedis.setex(key, seconds, value);
			}
		});
	}

	public Long expire(final String key, final int seconds) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.expire(key, seconds);
			}
		});
	}

	public Long incr(final String key) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.incr(key);
			}
		});
	}

	public Long decr(final String key) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.decr(key);
			}
		});
	}

	public Long hset(final String key, final String field, final String value) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.hset(key, field, value);
			}
		});
	}

	public String hget(final String key, final String field) {
		return execute(key, new HashRedisExecutor<String>() {
			@Override
			public String execute(Jedis jedis) {
				return jedis.hget(key, field);
			}
		});
	}

	public String hmset(final String key, final Map<String, String> hash) {
		return execute(key, new HashRedisExecutor<String>() {
			@Override
			public String execute(Jedis jedis) {
				return jedis.hmset(key, hash);
			}
		});
	}

	public List<String> hmget(final String key, final String... fields) {
		return execute(key, new HashRedisExecutor<List<String>>() {
			@Override
			public List<String> execute(Jedis jedis) {
				return jedis.hmget(key, fields);
			}
		});
	}

	public Long del(final String key) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.del(key);
			}
		});
	}

	public Map<String, String> hgetAll(final String key) {
		return execute(key, new HashRedisExecutor<Map<String, String>>() {
			@Override
			public Map<String, String> execute(Jedis jedis) {
				return jedis.hgetAll(key);
			}
		});
	}

	public void destroy() {
		for (int i = 0; i < jedisPools.length; i++) {
			jedisPools[i].close();
		}
	}

	@Override
	public Object getObject(String key) {

		return execute(key, new HashRedisExecutor<Object>() {
			@Override
			public Object execute(Jedis jedis) {
//				byte[] genKey = genKey(key);
				byte[] bs = jedis.get(key.getBytes());
				Object res = null;
				if (bs != null) {
					res = SerializeUtil.deserialize(bs);
				}
				return res;

			}
		});
	}

	@Override
	public Object setObject(String key, Object value, int expireTime) {
		return execute(key, new HashRedisExecutor<Object>() {
			@Override
			public Object execute(Jedis jedis) {
				if (value == null) {
					return null;
				} else {
//					byte[] genKey = genKey(key);
					byte[] valueBytes = SerializeUtil.serialize(value);
					String rtn = jedis.set(key.getBytes(), valueBytes);
					jedis.expire(key.getBytes(), expireTime);
					return rtn;
				}
			}
		});
	}

	private static byte[] genKey(String key) {
		try {
//			return key.getBytes("utf-8");
			 return SerializeUtil.serialize(key);
		} catch (Exception e) {
			return key.getBytes();
		}
	}

	@Override
	public List<Object> keysObject(String key) {
		return execute(key, new HashRedisExecutor<List<Object>>() {
			@Override
			public List<Object> execute(Jedis jedis) {
				List<Object> listValue = new ArrayList<Object>();
				Set<String> keys = jedis.keys(key);
				for (String key : keys) {
					Object obj = getObject(key);
					listValue.add(obj);
				}
				return listValue;
			}
		});
	}

	@Override
	public ScanResult<String> scan(String cursor) {
		if (jedisPools.length > 1) {
			throw new NotImplementedException("HashRedisUtil.scan not implemented when jedisPools.length > 1");
		}
		return execute("", new HashRedisExecutor<ScanResult<String>>() {
			@Override
			public ScanResult<String> execute(Jedis jedis) {
				return jedis.scan(cursor);
			}
		});
	}

	@Override
	public ScanResult<String> scan(String cursor, ScanParams params) {
		if (jedisPools.length > 1) {
			throw new NotImplementedException("HashRedisUtil.scan not implemented when jedisPools.length > 1");
		}
		return execute("", new HashRedisExecutor<ScanResult<String>>() {
			@Override
			public ScanResult<String> execute(Jedis jedis) {
				return jedis.scan(cursor, params);
			}
		});
	}

	@Override
	public Long zadd(String key, double score, String member) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.zadd(key, score, member);
			}
		});
	}

	@Override
	public Long zadd(String key, Map<String, Double> scoreMembers) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.zadd(key, scoreMembers);
			}
		});
	}

	@Override
	public Double zincrby(String key, double score, String member) {
		return execute(key, new HashRedisExecutor<Double>() {
			@Override
			public Double execute(Jedis jedis) {
				return jedis.zincrby(key, score, member);
			}
		});
	}

	@Override
	public Long zcard(String key) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.zcard(key);
			}
		});
	}

	@Override
	public Long zinterstore(String dstkey, String... sets) {
		if (jedisPools.length > 1) {
			throw new NotImplementedException("HashRedisUtil.zinterstore not implemented when jedisPools.length > 1");
		}
		return execute(dstkey, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.zinterstore(dstkey, sets);
			}
		});
	}

	@Override
	public Set<String> zrange(String key, long start, long end) {
		return execute(key, new HashRedisExecutor<Set<String>>() {
			@Override
			public Set<String> execute(Jedis jedis) {
				return jedis.zrange(key, start, end);
			}
		});
	}

	@Override
	public Set<Tuple> zrangeWithScores(String key, long start, long end) {
		return execute(key, new HashRedisExecutor<Set<Tuple>>() {
			@Override
			public Set<Tuple> execute(Jedis jedis) {
				return jedis.zrangeWithScores(key, start, end);
			}
		});
	}

	@Override
	public Set<String> zrangeByScore(String key, double min, double max) {
		return execute(key, new HashRedisExecutor<Set<String>>() {
			@Override
			public Set<String> execute(Jedis jedis) {
				return jedis.zrangeByScore(key, min, max);
			}
		});
	}

	@Override
	public Set<Tuple> zrangeByScoreWithScores(String key, double min, double max) {
		return execute(key, new HashRedisExecutor<Set<Tuple>>() {
			@Override
			public Set<Tuple> execute(Jedis jedis) {
				return jedis.zrangeByScoreWithScores(key, min, max);
			}
		});
	}

	@Override
	public Long zrank(String key, String member) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.zrank(key, member);
			}
		});
	}

	@Override
	public Long zrem(String key, String... members) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.zrem(key, members);
			}
		});
	}

	@Override
	public Long zremrangeByRank(String key, long start, long end) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.zremrangeByRank(key, start, end);
			}
		});
	}

	@Override
	public Long zremrangeByScore(String key, double start, double end) {
		return execute(key, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.zremrangeByScore(key, start, end);
			}
		});
	}

	@Override
	public ScanResult<Tuple> zscan(String key, String cursor) {
		return execute(key, new HashRedisExecutor<ScanResult<Tuple>>() {
			@Override
			public ScanResult<Tuple> execute(Jedis jedis) {
				return jedis.zscan(key, cursor);
			}
		});
	}

	@Override
	public Double zscore(String key, String member) {
		return execute(key, new HashRedisExecutor<Double>() {
			@Override
			public Double execute(Jedis jedis) {
				return jedis.zscore(key, member);
			}
		});
	}

	@Override
	public Long zunionstore(String dstkey, String... sets) {
		if (jedisPools.length > 1) {
			throw new NotImplementedException("HashRedisUtil.zunionstore not implemented when jedisPools.length > 1");
		}
		return execute(dstkey, new HashRedisExecutor<Long>() {
			@Override
			public Long execute(Jedis jedis) {
				return jedis.zunionstore(dstkey, sets);
			}
		});
	}
}